package cn.edu.cug.cs.gtl.gsm.aof.hdfs;

import cn.edu.cug.cs.gtl.gsm.aof.AOFStorageManager;
import cn.edu.cug.cs.gtl.gsm.aof.DataFiles;
import cn.edu.cug.cs.gtl.gsm.aof.IndexFile;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.IOException;
import java.net.URI;

public class HDFSStorageManager extends AOFStorageManager {
//    protected static String uri = "hdfs://hecs-x-xlarge-2-linux-20201111193150:9000";

   protected static String uri = "hdfs://node1";
    protected FileSystem fs;

    /**
     * @param workDirectory
     * @param baseName
     * @param collectionSize
     * @param pageSize
     * @param overwrite
     */
    public HDFSStorageManager(FileSystem fs,String workDirectory, String baseName, long collectionSize, int pageSize, boolean overwrite) throws IOException {
        super(workDirectory, baseName, collectionSize, pageSize, overwrite);
        this.fs = fs;
        initial(overwrite);
    }



    @Override
    protected void closeStorages() throws IOException {
//        fs.close();
    }

    /**
     * 在workDirectory下创建baseName目录,并返回全路径目录，系统将在baseName目录下建立索引文件和数据文件
     *
     * @param workDirectory
     * @param baseName
     * @return
     * @throws IOException
     */
    @Override
    public boolean createDirectory(String workDirectory, String baseName) throws IOException {


        return createDirectory(fs, concatenate(workDirectory, baseName));
    }

    @Override
    public boolean existsFile(String fullPathName) throws IOException {
        return exitsFile(fs, fullPathName);
    }

    @Override
    public boolean existsDirectory(String fullPath) throws IOException {
        return existsDirectory(fs, fullPath);
    }

    @Override
    public boolean existsDirectory(String workDirectory, String baseName) throws IOException {
        return existsDirectory(fs, concatenate(workDirectory, baseName));
    }

    @Override
    public IndexFile createIndexFile() {
        return new HDFSIndexFile(fs, workDirectory, metaItem, pageMap);
    }

    @Override
    public DataFiles createDataFiles() {
        return new HDFSDataFiles(fs, workDirectory, metaItem, pageMap);
    }

    public static String concatenate(String workDirectory, String baseName) {
        if (workDirectory.endsWith("/")) {
            return (workDirectory + baseName);
        } else {
            return (workDirectory + "/" + baseName);
        }
    }

    public static String concatenate(String workDirectory, String baseName, String fileName) {
        String dir = concatenate(workDirectory, baseName);
        if (dir.endsWith("/")) {
            return (dir + fileName);
        } else {
            return (dir + "/" + fileName);
        }
    }

    public static boolean exitsFile(FileSystem fs, String fullPathName) throws IOException {
        return fs.exists(new Path(fullPathName));
    }

    public static boolean existsDirectory(FileSystem fs, String fullPath) throws IOException {
        return fs.isDirectory(new Path(fullPath));
    }
    public static boolean createDirectory(FileSystem fs, String fullPathName) throws IOException {
        boolean mkdirs = fs.mkdirs(new Path(fullPathName));
        return mkdirs;
    }

    public static FileSystem getConnection() throws IOException, InterruptedException {
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(uri), configuration, "root");
        return fs;
    }
}